package cn.virens.server.aio.channel.runnable;

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.virens.server.aio.channel.AioChannelSimple;
import cn.virens.server.common.WritePacket;

public class AioChannelWriteRunnable implements Runnable {
	private final Logger logger = LoggerFactory.getLogger(getClass());

	private final Queue<WritePacket> queue = new ConcurrentLinkedQueue<>();

	private final AioChannelSimple channel;

	public AioChannelWriteRunnable(AioChannelSimple channel) {
		this.channel = channel;
	}

	public void offer(WritePacket data) {
		this.queue.offer(data);
	}

	public void execute() {
		channel.getAcceptHandler().execute(this);
	}

	@Override
	public void run() {
		WritePacket packet = null;

		while ((packet = queue.poll()) != null) {
			if (channel.isRemove()) return;
			if (!channel.isOpen()) return;

			try {
				// 调用编码器对数据包进行编码 & 将数据包发送给终端
				ByteBuffer buffer = encode(packet);
				if (buffer != null && buffer.hasRemaining()) {
					writeToChannel(channel.getChannel(), buffer);
				} else {
					logger.error("发送失败:" + packet.toString());
				}
			} catch (Exception e) {
				logger.error(e.getMessage(), e);
			}
		}
	}

	private void writeToChannel(AsynchronousSocketChannel socketChannel, ByteBuffer buffer) {
		socketChannel.write(buffer, buffer, channel.getChannelWriteHandler());
	}

	private ByteBuffer encode(WritePacket packet) {
		return channel.getAioEncoder().encode(packet);
	}
}
